/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * 
 */
package eu.dicodeproject.analysis.restapi;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Properties;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.compress.archivers.ArchiveException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.oozie.client.OozieClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

import eu.dicodeproject.analysis.restapi.oozie.TwitterVectorExporter;

/**
 *
 */
@Controller
@RequestMapping("/twitter/")
public class TwitterController {

  private String hbaseTablename;
  private byte[] hbaseColumn;
  private HTablePoolHandler hTablePoolHandler;
  private byte[] hbaseColumnFamily;
  private TwitterVectorExporter exportService;
  private Properties exportProperties;
  private String twitterstreamTablename;

  /** Logger for this class. */
  private static final Logger log = LoggerFactory
      .getLogger(TwitterController.class);

  /**
   * @param topic
   * @param day
   * @param response
   * @throws IOException
   */
  @RequestMapping(value = "hashtag/{topic}_{day}", method = RequestMethod.GET)
  public void hashtag(@PathVariable String topic, @PathVariable String day,
      HttpServletResponse response) throws IOException {

    response.setContentType("application/json; charset=UTF-8");

    HTableInterface hTable = this.hTablePoolHandler
        .getHTable(this.hbaseTablename);

    Get get = new Get((topic + "_" + day).getBytes());
    get.addColumn(this.hbaseColumnFamily, this.hbaseColumn);
    Result result = hTable.get(get);
    final byte[] value = result.value();
    response.setContentLength(value.length);
    this.hTablePoolHandler.putHTable(hTable);
    OutputStream out = response.getOutputStream();
    out.write(value);
    out.flush();
    out.close();
  }

  /**
   * 
   * @param topic
   * @param minSupport
   * @param response
   * @throws IOException
   */

  @RequestMapping(value = "export/{topic}_{minSupport}", method = RequestMethod.GET)
  public void export(@PathVariable String topic,
      @PathVariable String minSupport, HttpServletResponse response) {

    InputStream in = null;
    OutputStream out = null;

    try {

      if (!this.validateParams(topic, minSupport)) {
        // response.setStatus(404); // No such resource
        response.sendError(404);
      } else {
        File zipFile = null;

        zipFile = exportService.getTwitterVectors(this.exportProperties);

        if (zipFile != null && zipFile.isFile()) {
          response.setContentType("application/zip; charset=UTF-8");
          response
              .setHeader("Content-Length", String.valueOf(zipFile.length()));
          response.setHeader("Content-Disposition", "attachment; filename=\""
              + zipFile.getName() + "\"");

          byte buff[] = new byte[8192];
          in = new FileInputStream(zipFile);
          out = response.getOutputStream();

          int i = 0;
          while ((i = in.read(buff)) > 0) {
            out.write(buff, 0, i);
            out.flush();
          }
          in.close();
          out.close();
        } else {
          // response.setStatus(403); // Error, no zip file
          response.sendError(403);
        }
      }
    } catch (OozieClientException e) {
      log.error("Error while exporting Tweets: " + e.getMessage());
    } catch (InterruptedException e) {
      log.error("Error while exporting Tweets: " + e.getMessage());
    } catch (ArchiveException e) {
      log.error("Error while writing Tweet archive file: " + e.getMessage());
    } catch (IOException e) {
      log.error("Error in Tweet Export: " + e.getMessage());
    } finally {
      if (out != null) {
        try {
          out.close();
        } catch (IOException e) {
          log.error("Error in Tweet Export: " + e.getMessage());
        }
      }
      if (in != null) {
        try {
          in.close();
        } catch (IOException e) {
          log.error("Error in Tweet Export: " + e.getMessage());
        }
      }
    }
  }

  /**
   * Check parameter AND add default parameters
   * 
   * @param topic
   * @param minSupport
   * @return
   */
  private boolean validateParams(String topic, String minSupport) {

    if (topic == null) {
      return false;
    }
    try {
      // check if minSupport is int
      Integer.parseInt(minSupport);
      this.exportProperties = new Properties();
      this.exportProperties.put("hbaseTable", twitterstreamTablename);
      this.exportProperties.put("topic", topic);
      this.exportProperties.put("minSupport", minSupport);
    } catch (NumberFormatException e) {
      return false;
    }
    return true;
  }

  /**
   * @param hbaseTablename
   */
  @Resource(name = "hbaseTablename")
  public void setHBaseTablename(String hbaseTablename) {
    this.hbaseTablename = hbaseTablename;
  }

  /**
   * @param hbaseColumnFamily
   */
  @Resource(name = "hbaseColumnFamily")
  public void setHBaseColumnFamily(String hbaseColumnFamily) {
    this.hbaseColumnFamily = hbaseColumnFamily.getBytes();
  }

  /**
   * @param hbaseColumn
   */
  @Resource(name = "hbaseColumn")
  public void setHBaseColumn(String hbaseColumn) {
    this.hbaseColumn = hbaseColumn.getBytes();
  }

  /**
   * 
   * @param twitterstreamTablename
   */
  @Resource(name = "twitterstreamTablename")
  public void setTwitterstreamTablename(String twitterstreamTablename) {
    this.twitterstreamTablename = twitterstreamTablename;
  }

  /**
   * @param handler
   */
  @Autowired
  public void setHTablePoolHandler(HTablePoolHandler handler) {
    this.hTablePoolHandler = handler;
  }

  /**
   * @param exportService
   */
  @Autowired
  public void setExportService(TwitterVectorExporter exportService) {
    this.exportService = exportService;
  }
}
